package com.smsc.headend.task.engine.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.smsc.headend.common.enums.DeviceType;
import com.smsc.headend.common.utils.RedisKeys;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.request.entity.LogComAccess;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.ConnectorDeviceNotification;
import com.smsc.headend.module.task.dto.DeviceStatusDTO;
import com.smsc.headend.task.engine.mapper.LogTerminalAccessMapper;
import com.smsc.headend.task.engine.po.DeviceDetail;
import com.smsc.headend.task.engine.registry.DataNotificationRegistry;
import com.smsc.headend.task.engine.service.DeviceNotificationService;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.push.DataNotifyService;
import com.smsc.headend.task.engine.service.push.ProtocolDataProcess;
import com.smsc.headend.task.engine.service.task.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * @author liangli
 * @date 2020/10/21
 */
@Service
@Slf4j
public class DeviceNotificationServiceImpl implements DeviceNotificationService {
    @Autowired
    RedisUtils redisUtils;
    @Autowired
    FeignAssetManagementService assetManagementService;
    @Autowired
    DataNotificationRegistry dataNotificationRegistry;
    @Autowired
    TaskService taskService;
    @Autowired
    AssetService assetService;
    @Autowired
    LogTerminalAccessMapper logTerminalAccessMapper;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    DataNotifyService dataNotifyService;

    /**
     * 获取规约对应的ServiceImpl,推送数据处理
     *
     * @param notification
     */
    @Override
    public void handleConnectorNotification(ConnectorDeviceNotification notification) {
        if (notification.getType() == ConnectorDeviceNotification.NotificationType.Event.getType()) {
            log.info("Device Event: notify={}", JSONUtil.toJsonStr(notification));
            handleDeviceEvent(notification);
        }
        //online/offline event
        else {
            handleDeviceOnlineStatus(notification);
        }
    }

    private void handleDeviceOnlineStatus(ConnectorDeviceNotification notification) {
        if (notification.getComId() == -1) {
            return;
        }
        log.info("com {} status {} tv {}", notification.getComId(), notification.getType(), notification.getTimestamp());
//        Long lastUpdateTime = Convert.toLong(redisUtils.get(RedisKeys.getComOnlineStatusUpdateTime(notification.getComId())));
//        if (lastUpdateTime != null && lastUpdateTime > notification.getTimestamp()) {
//            return;
//        }
        LogComAccess logTerminalAccess = new LogComAccess();
        logTerminalAccess.setAccessTv(notification.getTimestamp());
        logTerminalAccess.setAccessType(notification.getType());
        logTerminalAccess.setComId(notification.getComId());
        logTerminalAccessMapper.insert(logTerminalAccess);
//        int res = assetManagementService.updateComEntityStatus(notification.getComId(), notification.getType());
//        redisUtils.set(RedisKeys.getComOnlineStatusUpdateTime(notification.getComId()), notification.getTimestamp());
//        if (res == 0) {
//            log.error("update COM {} status  failed", notification.getComId());
//        }
        if (notification.getType() == ConnectorDeviceNotification.NotificationType.Online.getType()) {
//            kafkaTemplate.send(TaskKafkaTopic.TASK_RESUME, Convert.toStr(notification.getComId()), notification.getComId().toString().getBytes(StandardCharsets.UTF_8));
            processDeviceStatusToMdm(notification.getComId(), DeviceStatusDTO.DeviceStatus.Online.getCode(), notification.getTimestamp() == null? SystemClock.now() / 1000 : notification.getTimestamp());
        } else {
            taskService.interruptComTask(notification.getComId());
            if (notification.getType() == ConnectorDeviceNotification.NotificationType.Offline.getType()) {
                processDeviceStatusToMdm(notification.getComId(), DeviceStatusDTO.DeviceStatus.Offline.getCode(), notification.getTimestamp() == null? SystemClock.now() / 1000 : notification.getTimestamp());
            }
        }

    }

    private void handleDeviceEvent(ConnectorDeviceNotification notification) {
        Meter meter = StringUtils.isEmpty(notification.getDeviceAddress()) ? null : assetManagementService.getMeterByMfgSn(notification.getDeviceAddress());
        DeviceDetail deviceDetail = assetService.getDeviceDetail(notification.getComId(), meter == null ? null : meter.getMeterId());
//        ProtocolDataProcess protocolDataProcess = dataNotificationRegistry.getService(deviceDetail.getDeviceTypeGroup().getProtocolId().longValue());
//        protocolDataProcess.receiveNotificationData(deviceDetail.getDeviceTypeGroup().getDeviceType().longValue(), deviceDetail.getDeviceId()
//                , deviceDetail.getDeviceTypeGroup().getModelId(), notification.getData());
        dataNotifyService.receiveNotificationData(deviceDetail.getDeviceTypeGroup().getDeviceType().longValue(),
                deviceDetail,notification.getData());
    }

    /**
     * 处理终端在线状态给MDM
     * @param comId
     * @param onlineStatus
     */
    private void processDeviceStatusToMdm(Long comId, int onlineStatus, Long tv) {
        ComEntity comEntity = assetService.getComById(comId);
        if (comEntity != null) {
            List<DeviceStatusDTO> deviceStatusDTOS = new ArrayList<>();
            if (comEntity.getComDeviceType().equals(DeviceType.DCU.getId())) {
                // DCU类型时把DCU的状态发送给MDM
                DeviceStatusDTO deviceStatusDTO = new DeviceStatusDTO();
                deviceStatusDTO.setTv(tv);
                deviceStatusDTO.setDeviceType(DeviceType.DCU.getId());
                deviceStatusDTO.setDeviceStatus(onlineStatus);
                deviceStatusDTO.setDeviceId(comEntity.getDeviceId());
                if (StrUtil.isNotEmpty(comEntity.getIpAddress())) {
                    deviceStatusDTO.setNetAddr(comEntity.getIpAddress());
                }
                deviceStatusDTOS.add(deviceStatusDTO);
            } else if (comEntity.getComDeviceType().equals(DeviceType.GPRS.getId())){
                // P2P时只需要把表计状态发送给MDM
                DeviceStatusDTO deviceStatusDTO = new DeviceStatusDTO();
                deviceStatusDTO.setTv(tv);
                deviceStatusDTO.setDeviceType(DeviceType.METER.getId());
                deviceStatusDTO.setDeviceStatus(onlineStatus);
                deviceStatusDTO.setDeviceId(comEntity.getDeviceId());
                if (StrUtil.isNotEmpty(comEntity.getIpAddress())) {
                    deviceStatusDTO.setNetAddr(comEntity.getIpAddress());
                }
                deviceStatusDTOS.add(deviceStatusDTO);
            }
            if (CollUtil.isNotEmpty(deviceStatusDTOS)) {
                kafkaTemplate.send(TaskKafkaTopic.DEVICE_STATUS_CHANGE_TRANSFER, JSONUtil.toJsonStr(deviceStatusDTOS).getBytes(Charset.forName("utf-8")));
            }
        }
    }
}
